package com.mango.ch10;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;

import com.mango.HDFSTools.HDFSAPI;

import scala.Tuple2;
import scala.Tuple3;
import scala.Tuple7;

/**
 * 输入路径共享了FilmRecommondJob的路径资源
 * 
 * @author Mango
 *
 */
public class Spark_FilmRecommondJob {
	static Path inputPath = new Path("/FilmRecommondJob/input");// input
	static Path outPath = new Path("/FilmRecommondJob/spark_output"); // output

	@SuppressWarnings("serial")
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		// 检查路径问题
		chenckPath();
		SparkConf conf = new SparkConf();
		JavaSparkContext jsc = new JavaSparkContext(conf);
		// reade datas
		JavaRDD<String> lines = jsc.textFile(inputPath.toString());
		// create key&value for Job requried (movie,(user,pf))
		JavaPairRDD<String, Tuple2<String, Integer>> pairs = lines
				.mapToPair(new PairFunction<String, String, Tuple2<String, Integer>>() {

					@Override
					public Tuple2<String, Tuple2<String, Integer>> call(String t) throws Exception {
						// TODO Auto-generated method stub
						String[] tokens = t.split(" ");
						String user = tokens[0];
						String movie = tokens[1];
						Integer pf = Integer.valueOf(tokens[2]);
						return new Tuple2<String, Tuple2<String, Integer>>(movie,
								new Tuple2<String, Integer>(user, pf));
					}

				});
		pairs.saveAsTextFile(outPath.toString() + "/1");
		// 规约数据 按movie 对数据规约以统计 电影的评论人数
		// (movie1,[(user4,5), (user1,3), (user2,2), (user3,2)])
		JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> mvParis = pairs.groupByKey();
		mvParis.saveAsTextFile(outPath.toString() + "/2");
		// 统计电影的评论人数 组合新的键值对形式 格式为：(user1,(movie2,4,4))
		JavaPairRDD<String, Tuple3<String, Integer, Integer>> usersRDD = mvParis.flatMapToPair(
				new PairFlatMapFunction<Tuple2<String, Iterable<Tuple2<String, Integer>>>, String, Tuple3<String, Integer, Integer>>() {

					@Override
					public Iterator<Tuple2<String, Tuple3<String, Integer, Integer>>> call(
							Tuple2<String, Iterable<Tuple2<String, Integer>>> t) throws Exception {
						// TODO Auto-generated method stub
						List<Tuple2<String, Tuple3<String, Integer, Integer>>> list = new ArrayList<Tuple2<String, Tuple3<String, Integer, Integer>>>();
						String movie = t._1;
						int sumOfRate = 0;
						List<Tuple2<String, Integer>> tmpList = new ArrayList<Tuple2<String, Integer>>();
						for (Tuple2<String, Integer> tl : t._2) {
							sumOfRate += 1;
							tmpList.add(tl);
						}
						for (Tuple2<String, Integer> tl : tmpList) {
							String user = tl._1;
							int pf = tl._2();
							Tuple3<String, Integer, Integer> info = new Tuple3<String, Integer, Integer>(movie, pf,
									sumOfRate);
							Tuple2<String, Tuple3<String, Integer, Integer>> item = new Tuple2<String, Tuple3<String, Integer, Integer>>(
									user, info);
							list.add(item);
						}

						return list.iterator();
					}

				});
		usersRDD.saveAsTextFile(outPath.toString() + "/3");
		// 完成自连接(按键来匹配两个数据集,即组成所有的配对情况,然后过滤掉重复的就可以了)
		JavaPairRDD<String, Tuple2<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>>> joinRDD = usersRDD
				.join(usersRDD);
		joinRDD.saveAsTextFile(outPath.toString() + "/4");
		// 删除重复的movie1和movie2键值对
		JavaPairRDD<String, Tuple2<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>>> filteredRDD = joinRDD
				.filter(new Function<Tuple2<String, Tuple2<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>>>, Boolean>() {

					@Override
					public Boolean call(
							Tuple2<String, Tuple2<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>>> v1)
							throws Exception {
						// TODO Auto-generated method stub
						// 删除所有重复项，以确保所有的movie1和movie2都有movie1<movie2
						Tuple3<String, Integer, Integer> m1 = v1._2._1;
						Tuple3<String, Integer, Integer> m2 = v1._2._2;
						String m1Name = m1._1();
						String m2Name = m2._1();
						if (m1Name.compareTo(m2Name) < 0) {
							return true;
						} else
							return false;
					}
				});
		filteredRDD.saveAsTextFile(outPath.toString() + "/5");
		// 生成所有movie1和movie2的组合(注意前面的自连接，然后又过滤了一次，就已经生成了movie1和movie2的组合所有组合了，注意不唯一，但一定是m1.name<m2.name这种格式的(只是去掉了同一组数据进行自连接时如：a,b
		// 会产生a b 和b a) b>a的情况，因为我们要让其数据的生成保持一种规则，只要其中一种的数据即可)
		// 此处的生成所有m1和m2的组合是将前面的结果(user3,((movie1,2,4),(movie2,3,4)))
		// 去掉user字段然后新添几个字段即可得到业务所需格式
		JavaPairRDD<Tuple2<String, String>, Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>> moviePairs = filteredRDD
				.mapToPair(
						new PairFunction<Tuple2<String, Tuple2<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>>>, Tuple2<String, String>, Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>>() {

							@Override
							public Tuple2<Tuple2<String, String>, Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>> call(
									Tuple2<String, Tuple2<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>>> t)
									throws Exception {
								// String user=t._1//将其去除
								Tuple3<String, Integer, Integer> m1 = t._2._1;
								Tuple3<String, Integer, Integer> m2 = t._2._2;
								Tuple2<String, String> m1m2Key = new Tuple2<String, String>(m1._1(), m2._1());
								int ratingProduct = m1._2() * m2._2();
								int rating1Squared = m1._2() * m1._2();
								int rating2Squared = m2._2() * m2._2();
								Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer> t7 = new Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>(
										m1._2(), m1._3(), m2._2(), m2._3(), ratingProduct, rating1Squared,
										rating2Squared);
								return new Tuple2<Tuple2<String, String>, Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>>(
										m1m2Key, t7);
							}
						});
		moviePairs.saveAsTextFile(outPath.toString() + "/6");
		// 由于键值对并不唯一，所以 对关联之后的电影进行规约分组
		JavaPairRDD<Tuple2<String, String>, Iterable<Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>>> grouped = moviePairs
				.groupByKey();
		grouped.saveAsTextFile(outPath.toString() + "/7");
		// 求关联度
		JavaPairRDD<Tuple2<String, String>, Tuple3<Double, Double, Double>> result = grouped.mapValues(
				new Function<Iterable<Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>>, Tuple3<Double, Double, Double>>() {

					@Override
					public Tuple3<Double, Double, Double> call(
							Iterable<Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>> v1)
							throws Exception {
						// TODO Auto-generated method stub
						return calculateCorrelations(v1);
					}

				});
		result.saveAsTextFile(outPath.toString() + "/8");
		jsc.close();
	}

	protected static Tuple3<Double, Double, Double> calculateCorrelations(
			Iterable<Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer>> v1) {
		// TODO Auto-generated method stub
		int groupSize = 0;
		int dotProduct = 0;
		int rating1Sum = 0;
		int rating2Sum = 0;
		int rating1NormSq = 0;
		int rating2NormSq = 0;
		int maxNumRaterS1 = 0;
		int maxNumRaterS2 = 0;
		for (Tuple7<Integer, Integer, Integer, Integer, Integer, Integer, Integer> t7 : v1) {
			groupSize += 1;
			dotProduct += t7._5();
			rating1Sum += t7._1();
			rating2Sum += t7._3();
			rating1NormSq = t7._6();
			rating2NormSq = t7._7();
			int numOfRaterS1 = t7._2();
			if (numOfRaterS1 > maxNumRaterS1)
				maxNumRaterS1 = numOfRaterS1;
			int numOfRaterS2 = t7._4();
			if (numOfRaterS2 > maxNumRaterS2)
				maxNumRaterS2 = numOfRaterS2;
		}
		double perason = calculatePerasonCorrelation(groupSize, dotProduct, rating1Sum, rating2Sum, rating1NormSq,
				rating2NormSq);
		double cosine = calculateCosinecorrelation(dotProduct, Math.sqrt(rating1NormSq), Math.sqrt(rating2NormSq));
		double jaccard = calculateJaccardCorrelation(groupSize, maxNumRaterS1, maxNumRaterS2);
		return new Tuple3<Double, Double, Double>(perason, cosine, jaccard);
	}

	private static double calculateJaccardCorrelation(int groupSize, int maxNumRaterS1, int maxNumRaterS2) {
		// TODO Auto-generated method stub
		double union = maxNumRaterS1 + maxNumRaterS2 - groupSize;
		return groupSize / union;
	}

	private static double calculateCosinecorrelation(int dotProduct, double sqrt, double sqrt2) {
		// TODO Auto-generated method stub
		return dotProduct / (sqrt * sqrt2);
	}

	private static double calculatePerasonCorrelation(int groupSize, int dotProduct, int rating1Sum, int rating2Sum,
			int rating1NormSq, int rating2NormSq) {
		// TODO Auto-generated method stub
		double numerator = groupSize * dotProduct - rating1Sum * rating2Sum;
		double denominator = Math.sqrt(groupSize * rating2NormSq - rating2Sum * rating2Sum)
				* Math.sqrt(groupSize * rating1NormSq - rating1Sum * rating1Sum);
		return numerator / denominator;
	}

	static void chenckPath() {
		// 创建输入文件夹
		Configuration conf = new Configuration();
		try {
			HDFSAPI hdfs = new HDFSAPI(conf);
			// hdfs.createDirectory(transactions);
			// 先删除已经有的输出文件夹
			hdfs.delDirectory(outPath);
			hdfs.closeFilseSystem();
			// hdfs.orpOver();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			System.out.println("----------文件操作失败");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	protected static Tuple2<Long, Long> buildSortedTuple(Long p1, Long p2) {
		// TODO Auto-generated method stub
		if (p1 > p2) {
			return new Tuple2<Long, Long>(p2, p1);
		} else
			return new Tuple2<Long, Long>(p1, p2);
	}
}
